package cn.itcast.flink.savepoint;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class SavepointDemo {
    public static void main(String[] args) throws Exception {
        /**
         * 实现步骤：
         * 1）初始化flink流处理的运行环境
         * 2）开启checkpoint
         * 3）指定数据源
         * 4）对字符串进行空格拆分，然后每个单词记一次数
         * 5）对每个单词进行分组聚合操作
         * 6) 打印测试
         * 7）执行任务，递交作业
         */

        //TODO 1）初始化flink流处理的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //TODO 2）开启checkpoint
        //周期性的生成barrier（栅栏），默认情况下checkpoint是没有开启的
        env.enableCheckpointing(5000L);
        //设置checkpoint的超时时间
        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
        //设置同一个时间只能有一个栅栏在运行
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
        // 设置checkpoint的执行模式，最多执行一次或者至少执行一次
        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        //Checkpointing最小时间间隔
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(500);
        //指定checkpoint的存储位置
        env.getCheckpointConfig().setCheckpointStorage("hdfs://node1:8020/flink-checkpoints");

        //TODO 取消作业的时候，上一次成功checkpoint的结果，被删除了！意味着不能将上次执行累加的结果无法恢复，因此我们希望取消作业的时候，不要删除已经checkpoit成功的历史数据
        //ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION: 当作业被取消时，保留外部的checkpoint。注意，在此情况下，您必须手动清理checkpoint状态。
        //ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION: 当作业被取消时，删除外部化的checkpoint。只有当作业失败时，检查点状态才可用。
        env.getCheckpointConfig()
                .enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //TODO 3）指定数据源
        DataStreamSource<String> socketTextStream = env.socketTextStream("node1", 9999);

        //TODO 4）对字符串进行空格拆分，然后每个单词记一次数
        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = socketTextStream.flatMap(new FlatMapFunction<String, Tuple2<String, Long>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Long>> collector) throws Exception {
                String[] words = line.split(" ");
                for (String word : words) {
                    collector.collect(Tuple2.of(word, 1L));
                }
            }
        });

        //TODO 5）对每个单词进行分组聚合操作
        SingleOutputStreamOperator<Tuple2<String, Long>> sumed = wordAndOne.keyBy(t->t.f0).sum(1);

        //TODO 6) 打印测试
        sumed.print();

        //TODO 7）执行任务，递交作业
        env.execute();
    }
}